package cc.toprank.syncdatatask.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;

@Component
public class CanalClient extends AbstractCanalClient {
    @Value("${canal.server.port}")
    String canalServerPort ;
    @Value("${canal.instance.name}")
    String canalInstanceName;
    @Value("${canal.server.ip}")
    String canalServerIp ;
    @Value("${no_sync}")
    String NO_SYNC;
    @Value("${delete_sync}")
    String DELETE_SYNC;


    public CanalClient(){
        super();
    }

    public CanalClient(String destination) {
        super(destination);
    }

    public void startCanal(RabbitTemplate rabbitTemplate)  {
        // 根据ip，直接创建链接，无HA的功能AddressUtils.getHostIp()
        String destination = "example";
        // 创建链接
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress(canalServerIp, Integer.parseInt(canalServerPort)), canalInstanceName,
                "", "");

        final CanalClient canalClient = new CanalClient(destination);
        canalClient.setConnector(connector);
        canalClient.setRabbitTemplate(rabbitTemplate);
        canalClient.setNO_SYNC(NO_SYNC);
        canalClient.setDELETE_SYNC(DELETE_SYNC);
        canalClient.start();
        Runtime.getRuntime().addShutdownHook(new Thread() {

            public void run() {
                try {
                    logger.info("## stop the canal client");
                    canalClient.stop();
                } catch (Throwable e) {
                    logger.warn("##something goes wrong when stopping canal:\n{}", ExceptionUtils.getFullStackTrace(e));
                } finally {
                    logger.info("## canal client is down.");
                }
            }

        });
    }

}
